package com.atguigu.flink.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by Smexy on 2023/2/1
 *
 *  算子的处理函数中定义的非只读变量，称为状态！
 *
 *      之前声明的算子的处理函数中定义的非只读变量，称为 RawState(原始状态)，需要你自己去维护！
 *          一旦程序故障，重启后，状态就丢失了！
 *
 *      现在要学的是Flink提供的状态，称为ManageState。由Flink自动备份，自动分配，自动伸缩。
 *           一旦程序故障，重启后，状态和故障之前是一致的，可以无缝对接，继续处理！
 *
 *   ---------------------
 *      Task 故障： 某个算子所在的Task，在处理时，出异常。
 *      Job  故障:  整个Job停止运行，如果是per job模式，app停止运行。
 *                      通常情况会由JM挂掉引起，或遇到了无法处理的异常！
 *
 *      JobException: Recovery(恢复) is suppressed(抑制) by NoRestartBackoffTimeStrategy
 *                      虽然故障了，但是尝试恢复，但是失败了。被NoRestartBackoffTimeStrategy 抑制。
 *
 *    ---------------------
 *          Flink提供的 ManageState 就是 集合或变量！帮你存储变量！
 *
 */
public class Demo1_State
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        //指定 Task failover(故障转移)的策略   当Task故障时，最多尝试重启3次，每次间隔3000ms
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000));

        //开启备份，默认每间隔2000ms，帮你做一次备份。  无限重启
        env.enableCheckpointing(500);

                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new MyMap())
                   .addSink(new SinkFunction<String>()
                   {
                       @Override
                       public void invoke(String value, Context context) throws Exception {
                           if (value.contains("x")){
                               throw new RuntimeException("出异常了...");
                           }
                           System.out.println(value);
                       }
                   });



                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }

    //累积字符串
    public static class MyMap implements MapFunction<String,String>, CheckpointedFunction
    {

        //RawState
        //List<String> strs = new ArrayList<>();

        //用这个保存数据，当集合用
        ListState<String> strState ;

        @Override
        public String map(String value) throws Exception {
            //加入状态
            strState.add(value);
            //从状态取
            List<String> strs = MyUtil.parseList(strState.get());
            return strs.toString();
        }

        //把状态用快照形式备份  2000ms调用一次
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            //System.out.println("MyMap.snapshotState");
            //在备份时，加入要备份的状态信息
            //strState.addAll(strs);
        }

        //在Task创建时，从备份中读取快照恢复状态    执行1次
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("MyMap.initializeState");
            //从备份中恢复状态  先获取OperatorState的存储，从中根据strs这个id获取一个ListState
            // ListState： 均匀分配
            //strState = context.getOperatorStateStore().getListState(new ListStateDescriptor<String>("strs",String.class));
            /*
                    UnionListState: 全量复制

                    只在KafkaSource中会使用。

                        一个subTask读取kafka主题的一个分区。
                            举例： 当前2个subTask
                                source1 :  读取 topicA-1 ,topicA-0
                                        在source的状态中会记录读取的偏移量
                                                topicA-1-20
                                                topicA-0-10

                                source2 :  读取 topicA-2 ,topicA-3
                                                topicA-2-30
                                                topicA-3-40

                            程序故障，重启后，需要reblance(重新分配)
                                  source1 :  读取 topicA-2 ,topicA-0
                                        恢复时，KafkaSource使用UnionListState，将状态的全部分给每个subTask
                                            topicA-1-20,topicA-0-10,topicA-2-30,topicA-3-40
                                  source2 :  读取 topicA-1 ,topicA-3
                                            topicA-1-20,topicA-0-10,topicA-2-30,topicA-3-40


             */
            strState = context.getOperatorStateStore().getUnionListState(new ListStateDescriptor<String>("strs",String.class));
        }
    }
}
